[feat][pip][consumer] Add Pause/Resume/Paused to Consumer by PavelZeger · Pull Request #1507 · apache/pulsar-client-go

PavelZeger · 2026-05-30T13:07:18Z

Motivation

The Java client exposes pause() / resume() on every Consumer; the Go client has no equivalent. While paused, a consumer should stop issuing flow permits to the broker so no new
messages are delivered, while already-buffered, in-flight, and previously-granted messages remain available via Receive() / Chan() and ack/nack keep working.

Pause/resume is the standard backpressure pattern for when a downstream is temporarily unavailable (database failover, a "halt ingestion" feature flag, orderly shutdown that finishes in-flight work before closing). The only workaround today is closing and re-subscribing, which is heavyweight (re-resolves the topic, redelivers unacked messages, and can rebalance other consumers in KeyShared/Failover).

The semantics were reviewed and agreed in #1504.

Modifications

Added Pause(), Resume(), and Paused() to the Consumer interface pulsar/consumer.go), with GoDoc matching the Java semantics.
partitionConsumer (pulsar/consumer_partition.go):
Added a paused flag and gated permit flow in flowIfNeed — the single place that returns permits to the broker.
When paused, the dispatcher connect/reconnect branch stashes the initial grant into availablePermits instead of sending it, so reconnecting while paused does not resume delivery.
resume() flushes only the permits actually owed (via flowIfNeed), mirroring Java's increaseAvailablePermits(cnx, 0), rather than granting a fresh full batch (which would over-grant after a mid-stream pause).
Wired the API through all four Consumer implementations: consumer, zeroQueueConsumer, multiTopicConsumer, and regexConsumer. Paused() is reported at the top level ("Pause() called more recently than Resume()").
Propagated the paused state to dynamically created children: new partitions (internalTopicSubscribeToPartitions) and newly discovered regex topics (regexConsumer.subscribe) inherit the parent's paused state at creation.
For the zero-queue consumer, Receive() is unchanged: the existing permit is gated while paused, so Receive(ctx) blocks until Resume() (or the context is cancelled).

Deviations from the original PIP

The observable behavior agreed in #1504 is unchanged; four implementation details differ from the issue's wording. Resume() flushes only the permits actually owed (and the dispatcher stashes the initial grant while paused) instead of sending a full currentQueueSize batch, which would over-grant after a mid-stream pause. The zero-queue consumer reuses the same flow gate instead of a new resumedCh. The getter is named Paused() rather than IsPaused(). BatchReceive is out of scope, since the Go client doesn't have that method yet.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Added TestConsumerPauseResume, TestPartitionedConsumerPauseResume, TestZeroQueueConsumerPauseResume, and TestRegexConsumerPauseResumeInheritsNewTopics in pulsar/consumer_test.go. They verify that a paused consumer stops delivery before draining the full backlog, that Resume() delivers the remainder with no message loss, that Pause()/Resume() are idempotent, that Paused() reflects state, that a newly discovered regex topic inherits the paused state, and that a zero-queue Receive() blocks while paused and returns after resume.
Verified locally against Apache Pulsar 4.0.3 standalone (all four tests pass, repeated runs stable);

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API: yes — adds Pause(), Resume(), and Paused() to the Consumer interface
The schema: no
The default values of configurations: no
The wire protocol: no (uses the existing CommandFlow; pausing simply withholds it)

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? GoDocs (doc comments on the new Consumer interface methods)

PavelZeger · 2026-06-03T19:33:00Z

Hi @nodece and @crossoverJie . Appreciate for the review

Copilot

Pull request overview

This PR adds a pause/resume backpressure API to the Go Consumer to match the Java client’s semantics: when paused, the consumer stops issuing flow permits so the broker stops delivering new messages while already-buffered/in-flight messages remain consumable and ack/nack continues to work.

Changes:

Added Pause(), Resume(), and Paused() to the public Consumer interface with detailed GoDoc describing semantics across subscription types.
Implemented paused state propagation across consumer implementations, and gated flow-permit issuance in partitionConsumer (including reconnect initial permit handling).
Added integration tests covering pause/resume behavior for standard, partitioned, zero-queue, and regex consumers.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
pulsar/consumer.go	Extends the public `Consumer` interface with Pause/Resume/Paused and documents expected semantics.
pulsar/consumer_impl.go	Wires Pause/Resume/Paused through the main consumer and ensures newly discovered partitions inherit paused state.
pulsar/consumer_partition.go	Adds per-partition paused flag and gates flow-permit issuance (including reconnect initial grants).
pulsar/consumer_multitopic.go	Propagates Pause/Resume/Paused across multi-topic consumers.
pulsar/consumer_regex.go	Adds pause state to regex consumer and ensures newly discovered topics inherit paused state.
pulsar/consumer_zero_queue.go	Implements Pause/Resume/Paused for zero-queue consumer via its underlying partitionConsumer.
pulsar/consumer_test.go	Adds pause/resume tests and helper drain functions used by the new test cases.
pulsar/internal/pulsartracing/consumer_interceptor_test.go	Updates a mock consumer to satisfy the expanded `Consumer` interface.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

# Conflicts: # pulsar/consumer_test.go

PavelZeger · 2026-06-12T13:36:02Z

Additional note: there is a known edge case in the partitioned/regex path. If a new partition (or regex topic) is discovered at the exact moment someone calls Resume(), that new child can end up paused while Paused() says false. It fixes itself on the next Pause()/Resume(), and the timing window is tiny.
I'm leaving it as-is here on purpose. #1508 (StartPaused) fixes it at the root by creating new children already paused instead of pausing them after the fact - same as the Java client. A lock here would just get undone once #1508 will be executed.

PavelZeger · 2026-10-30T13:35:46Z

Further this conversation I would like to mark this PR as active: https://lists.apache.org/thread/cgy7w0qgnq8s7j0ykyzgrn3hj4nb67wf.

# Conflicts: # pulsar/consumer_test.go

nodece

I did a deep pass on the concurrency aspects of this change (plus ran the four new tests and the consumer regression suites against a live broker — all green, and I additionally probed the reconnect-while-paused scenario 3x: 0 messages delivered while paused, exact remainder after Resume(), so the dispatcher stash works).

What I verified as correct

flowIfNeed is indeed the single permit-return path (checked every availablePermits.add/inc call site); the dispatcher initial grant is the one exception and is correctly stashed.
flush() and flowIfNeed() both claim permits via permits.CompareAndSwap(current, 0) — concurrent execution can neither double-send nor lose permits.
resume()'s CompareAndSwap(true, false) guards against double-flush, and the re-check of paused inside flush() handles the resume→pause interleaving correctly.
internalFlow from an application goroutine is safe — the zero-queue Receive path already does this today.

Nit (non-blocking): top-level Pause/Resume are multi-step and not serialized

func (c *consumer) Pause() {
	c.paused.Store(true)
	for _, pc := range c.partitionConsumers() { pc.pause() }
}

Concurrent Pause()/Resume() on the same consumer can interleave as: top-level flag false, partition flag true, flush skipped — i.e. Paused() reports resumed while delivery is actually paused. It self-heals on the next call and requires the application to race the two control calls, so I don't think it should block this PR (the Java client is similarly unsynchronized), but if you want to harden it:

CompareAndSwap in pause() would not help — pause() has no side effect to guard, and CAS cannot atomize the "flag + N partitions" multi-step operation.
A channel/actor loop would serialize it but costs a goroutine per consumer, close semantics, and a done-ack to preserve the current "Pause() returns → paused" guarantee.

The proportionate fix is a control-plane mutex that serializes the writers while keeping the readers lock-free:

type consumer struct {
	paused  atomic.Bool // readers (flowIfNeed, Paused()) stay lock-free
	pauseMu sync.Mutex  // serializes Pause/Resume writers
}

func (c *consumer) Pause() {
	c.pauseMu.Lock()
	defer c.pauseMu.Unlock()
	c.paused.Store(true)
	for _, pc := range c.partitionConsumers() { pc.pause() }
}
// Resume() symmetric

Taking the same pauseMu in internalTopicSubscribeToPartitions would additionally close the partition-discovery inheritance race (discovery checks c.paused per new partition on a background goroutine, which can also race a concurrent Pause()).

Test suggestion (also non-blocking)

The current tests don't cover reconnecting while paused — the most subtle path of the change (the stash). A cheap test: pause, pc._getConn().Close(), assert the backlog does not fully drain while paused, then Resume() and drain the exact remainder. Happy to contribute the probe I used if useful.

PavelZeger · 2026-11-07T10:28:07Z

Thank you, @nodece , for the review. Do you prefer all fixes to push into this branch or approve now and I'll open a new dedicated branch with fixes?

PavelZeger added 3 commits May 30, 2026 16:01

[feat][pip] Add Pause and Resume to Consumer

811a816

[feat][pip] Add Pause and Resume to Consumer

6f58639

[feat][pip] Add Pause and Resume to Consumer

5c44d0a

PavelZeger mentioned this pull request Jun 2, 2026

[PIP] Add Consumer.RedeliverUnacknowledgedMessages() #1510

Open

crossoverJie requested a review from Copilot June 8, 2026 02:40

Copilot started reviewing on behalf of crossoverJie June 8, 2026 02:41

View session

Copilot AI reviewed Jun 8, 2026

View reviewed changes

Comment thread pulsar/consumer_partition.go Resolved

Comment thread pulsar/consumer_test.go Resolved

Comment thread pulsar/consumer_test.go Resolved

PavelZeger and others added 3 commits June 8, 2026 22:34

Test fix

5fc40cf

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

[feat][pip] Review's fixes

9aa0791

[feat][pip] Review's fixes

7673135

PavelZeger mentioned this pull request Jun 8, 2026

[improve][client] Make Consumer.resume() eagerly top up the broker's permit window apache/pulsar#25978

Closed

2 tasks done

crossoverJie requested a review from Copilot June 9, 2026 02:56

Copilot started reviewing on behalf of crossoverJie June 9, 2026 02:56

View session

Copilot AI reviewed Jun 9, 2026

View reviewed changes

Comment thread pulsar/consumer_partition.go Resolved

Comment thread pulsar/consumer_test.go Resolved

Comment thread pulsar/consumer_regex.go Resolved

PavelZeger added 2 commits June 12, 2026 15:46

[feat][pip] Review's fixes

4bc453f

Merge branch 'master' into pip-1504-add-pause-resume-to-consumer

110929c

# Conflicts: # pulsar/consumer_test.go

Merge branch 'master' into pip-1504-add-pause-resume-to-consumer

f9ef383

# Conflicts: # pulsar/consumer_test.go

nodece reviewed Nov 2, 2026

View reviewed changes

Navigation Menu

Repository navigation

[feat][pip][consumer] Add Pause/Resume/Paused to Consumer - #1507

[feat][pip][consumer] Add Pause/Resume/Paused to Consumer#1507
PavelZeger wants to merge 9 commits into
apache:masterfrom
PavelZeger:pip-1504-add-pause-resume-to-consumer

PavelZeger commented May 30, 2026 •

edited

Loading

Uh oh!

PavelZeger commented Jun 3, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

PavelZeger commented Jun 12, 2026 •

edited

Loading

Uh oh!

PavelZeger commented Oct 30, 2026

Uh oh!

nodece left a comment

Uh oh!

PavelZeger commented Nov 7, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

PavelZeger commented May 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Deviations from the original PIP

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

PavelZeger commented Jun 3, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

PavelZeger commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

PavelZeger commented Oct 30, 2026

Uh oh!

nodece left a comment

Choose a reason for hiding this comment

What I verified as correct

Nit (non-blocking): top-level Pause/Resume are multi-step and not serialized

Test suggestion (also non-blocking)

Uh oh!

PavelZeger commented Nov 7, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

PavelZeger commented May 30, 2026 •

edited

Loading

PavelZeger commented Jun 12, 2026 •

edited

Loading